package com.boarsoft.dubbo;

import com.boarsoft.rpc.adapter.RpcProtocolAdapter;
import com.boarsoft.rpc.bean.RpcFaceConfig;
import com.boarsoft.rpc.bean.RpcMethodConfig;
import com.boarsoft.rpc.bean.RpcReferenceConfig;
import com.boarsoft.rpc.bean.RpcServiceConfig;
import com.boarsoft.rpc.core.RpcContext;
import com.boarsoft.rpc.core.RpcCore;
import com.boarsoft.rpc.spy.RpcSvcSpy;
import org.apache.dubbo.config.*;
import org.apache.dubbo.rpc.service.GenericService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.ApplicationContext;

import javax.annotation.PostConstruct;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;

public class DubboAdapter implements RpcProtocolAdapter {
	private static final Logger log = LoggerFactory.getLogger(DubboAdapter.class);
	/** */
	@Autowired
	protected DefaultListableBeanFactory proxyFactory;
	@Autowired
	protected ApplicationContext applicationContext;

	@Autowired(required = false)
	protected RpcSvcSpy svcSpy;
	@Autowired
	protected RpcContext rpcContext;

	protected ApplicationConfig application;
	protected ServiceConfig serverConfig;
	protected RegistryConfig registryConfig;

	protected String appName;
	protected String regAddress;
	protected int localPort = 12200;
	protected String rpcProtocol = "dubbo";
	protected String regProtocol = "zookeeper";
	protected int registTimeout = 3000;
	protected int addressWait;

	protected Map<String, Future<Object>> referMap = new ConcurrentHashMap<>();
	protected Map<String, CountDownLatch> locksMap = new ConcurrentHashMap<>();

	@PostConstruct
	public void init() {
		application = new ApplicationConfig(appName);
		registryConfig = new RegistryConfig();
		registryConfig.setAddress(regAddress);
	}

	@SuppressWarnings("unchecked")
	@Override
	public boolean registService(RpcServiceConfig sc, boolean export) throws ClassNotFoundException {
		// 构建 ProviderConfig
		Object ref = applicationContext.getBean(sc.getRef());
		ServiceConfig<Object> pc = new ServiceConfig<>();
		pc.setRegistry(registryConfig);
		pc.setApplication(application);
		pc.setRef(ref);
		pc.setInterface(sc.getInterfaceClazz().getName());
		// .setGroup(sc.getGroup())
		// .setVersion(sc.getVersion());
		pc.setTimeout(sc.getTimeout());//
		pc.setMethods(new ArrayList<>());//
		ProtocolConfig protocol = new ProtocolConfig();
		protocol.setName(rpcProtocol);
		protocol.setPort(localPort);
		pc.setProtocol(protocol);
		Map<String,String> params=new HashMap<>();
		params.put("sign", sc.getSign());
		params.put("tag", System.getProperty("rpc.tag", "Z"));
		pc.setParameters(params);// 传递serviceKey
		pc.setRegister(true);
		// 登记方法配置
		Map<String, RpcMethodConfig> mmcMap = sc.getMethodConfigMap();
		//this.registMethods(mmcMap, pc.getMethods());
		//Registry reg = ExtensionLoader.getExtensionLoader(RegistryFactory.class).getAdaptiveExtension().getRegistry(URL.valueOf(regAddress));
		// 通过自定义参数传递方法relativeId
		for (RpcMethodConfig mm : mmcMap.values()) {
			pc.getParameters().put(mm.getKey(), new StringBuilder("relativeId:")//
					.append(mm.getRelativeId()).toString());
		}
		// export
		if (export) {
			log.info("Export service {}", sc.getSign());
			pc.export();
		} else {
			// 只注册，不Export
			log.info("Regist service {}", sc.getSign());
			//reg.regist(pc);
		}
		return true;
	}

	protected ReferenceConfig<?> buildConsumerConfig(RpcReferenceConfig rc) {
		// 构建 ConsumerConfig
		ReferenceConfig<?> cc = new ReferenceConfig<>();
		cc.setApplication(application);
		cc.setRegistry(registryConfig);
		cc.setInterface(rc.getInterfaceName());
		cc.setGeneric(rc.isGeneralized());
		cc.setId(rc.getId());
				// .setGroup(rc.getGroup())
				// .setVersion(rc.getVersion())
		cc.setCheck(false);
		cc.setTimeout(rc.getTimeout());
		cc.setUrl(rc.getDirect());
		cc.setMethods(new ArrayList<>());

		Map<String,String> params=new HashMap<>();
		params.put("tag", System.getProperty("rpc.tag", "Z"));
		cc.setParameters(params);
		return cc;
	}

	@SuppressWarnings("unchecked")
	public <T> T referReference(RpcReferenceConfig rc, Class<T> clazz)
			throws ClassNotFoundException, InterruptedException, ExecutionException, TimeoutException {
		Future<Object> ft = this.registReference(rc);
		return (T) ft.get(registTimeout, TimeUnit.MILLISECONDS);
	}

	@Override
	public Future<Object> registReference(RpcReferenceConfig rc) throws ClassNotFoundException, InterruptedException {
		String key = rc.getSign();
		Future<Object> ro = referMap.get(key);
		if (ro != null) {
			return ro;
		}
		synchronized (referMap) {
			ro = referMap.get(key);
			if (ro != null) {
				return ro;
			}
			ReferenceConfig<?> cc = this.buildConsumerConfig(rc);
			cc.setProtocol(rpcProtocol);
			ExecutorService es = RpcCore.getCurrentInstance().getThreadPool();
			if (rc.isGeneralized()) {
				log.info("Create generic refer of {}", rc);
				ro = es.submit(new Callable<Object>() {
					@Override
					public Object call() throws Exception {
						return cc.get();
					}
				});
			} else {
				log.info("Create reference factory for {}", rc);
				// 设置通信协议
				BeanDefinitionBuilder bdb = BeanDefinitionBuilder//
						.genericBeanDefinition(DubboReferenceFactory.class);
				// bdb.addPropertyValue("refObject", ref);
				bdb.addPropertyValue("referenceConfig", cc);
				bdb.addPropertyValue("refClazz", rc.getInterfaceClazz());
				AbstractBeanDefinition abd = bdb.getRawBeanDefinition();
				proxyFactory.registerBeanDefinition(rc.getId(), abd);
				// 异步获取此bean，触发refer动作，避免首次调用高耗时
				ro = es.submit(new Callable<Object>() {
					@Override
					public Object call() throws Exception {
						return applicationContext.getBean(rc.getId(), rc.getInterfaceClazz());
					}
				});
			}
			referMap.put(key, ro);
			return ro;
		}
	}

	@Override
	public boolean subscribe(RpcReferenceConfig rc, boolean await) throws ClassNotFoundException, InterruptedException {
		String key = rc.getSign();
		if (locksMap.containsKey(key)) {
			return true;
		}
		synchronized (locksMap) {
			if (locksMap.containsKey(key)) {
				return true;
			}
			locksMap.put(key, new CountDownLatch(1));
			ReferenceConfig<?> cc = this.buildConsumerConfig(rc);
			cc.setProtocol(RpcFaceConfig.PROTOCOL_DEFAULT); // 订阅服务
			cc.setCheck(false);
			cc.get();
		}
		if (await) {
			locksMap.get(key).await(registTimeout, TimeUnit.MILLISECONDS);
		}
		return true;
	}


	@Override
	public Object invoke(RpcReferenceConfig rc, RpcMethodConfig rmc, Object[] args) throws InstantiationException,
			IllegalAccessException, ClassNotFoundException, InterruptedException, ExecutionException, TimeoutException {
		Future<Object> ft = this.registReference(rc);
		GenericService gs = (GenericService) ft.get(registTimeout, TimeUnit.MILLISECONDS);

		// 2.2 进行调用，不指定返回类型，返回结果类型为GenericObject
		Object[] ta = new Object[args.length];
		for (int i = 0; i < args.length; i++) {
			ta[i] = this.checkIn(args[i]);
		}
		String sign = rmc.getSign();
		int i = sign.indexOf("(");
		String mn = sign.substring(0, i);
		String pm = sign.substring(i + 1, sign.length() - 1);
		Object ro = gs.$invoke(mn, pm.split(","), ta);
		return this.checkOut(ro);

	}

	@SuppressWarnings("unchecked")
	public Object checkOut(Object x) throws InstantiationException, IllegalAccessException {
		if (x == null) {
			return null;
		}
		Class<?> clazz = x.getClass();
		if (clazz.isPrimitive() || x instanceof String || x instanceof Byte || x instanceof Short
				|| x instanceof Integer || x instanceof Long || x instanceof Float || x instanceof Double
				|| x instanceof Boolean || x instanceof BigDecimal) {
			return x;
		}
		if (clazz.isArray()) {
			Object[] sa = (Object[]) x;
			Object[] ta = new Object[sa.length];
			for (int i = 0; i < sa.length; i++) {
				ta[i] = this.checkOut(sa[i]);
			}
			return ta;
		}
		if (clazz.isAssignableFrom(Collection.class)) {
			Collection<Object> sa = (Collection<Object>) x;
			Collection<Object> ta = (Collection<Object>) sa.getClass().newInstance();
			for (Object o : sa) {
				ta.add(this.checkIn(o));
			}
			return ta;
		}
		return x;
	}

	@SuppressWarnings("unchecked")
	public Object checkIn(Object x) throws InstantiationException, IllegalAccessException {
		if (x == null) {
			return null;
		}
		Class<?> clazz = x.getClass();
		if (clazz.isPrimitive() || x instanceof String || x instanceof Byte || x instanceof Short
				|| x instanceof Integer || x instanceof Long || x instanceof Float || x instanceof Double
				|| x instanceof Boolean || x instanceof BigDecimal) {
			return x;
		}
		if (clazz.isArray()) {
			Object[] sa = (Object[]) x;
			Object[] ta = new Object[sa.length];
			for (int i = 0; i < sa.length; i++) {
				ta[i] = this.checkIn(sa[i]);
			}
			return ta;
		}
		if (clazz.isAssignableFrom(Collection.class)) {
			Collection<Object> sa = (Collection<Object>) x;
			Collection<Object> ta = (Collection<Object>) sa.getClass().newInstance();
			for (Object o : sa) {
				ta.add(this.checkIn(o));
			}
			return ta;
		}
		return x;
	}

	public String getRegAddress() {
		return regAddress;
	}

	public void setRegAddress(String regAddress) {
		this.regAddress = regAddress;
	}

	public int getLocalPort() {
		return localPort;
	}

	public void setLocalPort(int localPort) {
		this.localPort = localPort;
	}

	public String getAppName() {
		return appName;
	}

	public void setAppName(String appName) {
		this.appName = appName;
	}

	public String getRpcProtocol() {
		return rpcProtocol;
	}

	public void setRpcProtocol(String rpcProtocol) {
		this.rpcProtocol = rpcProtocol;
	}

	public String getRegProtocol() {
		return regProtocol;
	}

	public void setRegProtocol(String regProtocol) {
		this.regProtocol = regProtocol;
	}

	public int getRegistTimeout() {
		return registTimeout;
	}

	public void setRegistTimeout(int registTimeout) {
		this.registTimeout = registTimeout;
	}

	public int getAddressWait() {
		return addressWait;
	}

	public void setAddressWait(int addressWait) {
		this.addressWait = addressWait;
	}

}
